package com.rds.benchmark.thread;

import com.rds.benchmark.command.CommandOperator;
import com.rds.benchmark.config.ArgvsConfig;
import com.rds.benchmark.connector.Connector;
import com.rds.benchmark.operator.TypeOperator;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import org.redisson.api.RedissonClient;

import java.util.Date;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;

public class StableRunThread extends Thread{

    private TypeOperator[] operators;
    private CountDownLatch latch;

    public StableRunThread(String name) {
        super(name);
    }

    public void initParameter(TypeOperator[] operators, CountDownLatch latch) {
        this.operators = operators;
        this.latch = latch;
    }

    @Override
    public void run() {
        int idx = Integer.parseInt(Thread.currentThread().getName());
        int pxx_idx = idx % CommandOperator.PXX_MAX;
        String prefix = UUID.randomUUID().toString();
        String key = CommandOperator.createKey(prefix);
        TypeOperator operator = operators[idx];
        Object conn = null;
        try {
            //System.out.println("Thread-" + idx + " barriers waiting count:" + startBarrier.getNumberWaiting());
            Connector.startBarrier.await();

            boolean first = true;
            long stop = System.currentTimeMillis() + ArgvsConfig.milliTime;
            while (stop > System.currentTimeMillis()) {
                try {
                    conn = Connector.connector.getJedis();
                    if (conn instanceof RedissonClient) {
                        conn = ((RedissonClient) conn).getBucket(key);
                    } else if (conn instanceof StatefulConnection) {
                        if (conn instanceof StatefulRedisConnection)
                            conn = ((StatefulRedisConnection) conn).sync();
                        else if (conn instanceof StatefulRedisClusterConnection)
                            conn = ((StatefulRedisClusterConnection) conn).sync();
                    }
                    operator.init(conn, key, idx, null);
                    if (first) {
                        first = false;
                        //操作前清理之前的数据
                        operator.clear();
                    }

                    long sub_start = System.nanoTime();

                    operator.operate();

                    sub_start = (System.nanoTime() - sub_start) / 1000;
                    if (sub_start > 0) {
                        CommandOperator.PXXES[pxx_idx].set(sub_start);
                    }

                    if (CommandOperator.needPrint()) {
                        System.out.println(new Date() + " ok.");
                    }
                } catch (Throwable t) {
                    if (CommandOperator.needPrint()) {
                        System.out.println(new Date() + " Error: " + t.getMessage());
                    }
                } finally {
                    Connector.connector.close(conn);
                }
            }
        } catch (Throwable t) {
            System.err.println("Thread-" + idx + " run failed!");
//                        t.printStackTrace();
        } finally {
            operator.clear(); //清除测试数据
            operator.endTimes();
        }
        latch.countDown();
    }
}
